package org.kafka;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.Random;


public class kafka {



        public static void main(String[] args) {


            String topic = "stu3";
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "niit01:9092");
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer" );
            KafkaProducer<String, String> producer = new KafkaProducer<>(props);
            while (true) {
                int classId = (int) (Math.random() * 10) + 1;

                String name = generateName();  // 姓名为英文字母
                int gender = (int) (Math.random() * 2);  // 性别，1为男，0为女
                LocalDate birthDate = LocalDate.now().minusYears((int) (Math.random() * 20));
                String birthDateString = birthDate.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
                int studentId = (int) (Math.random() * 10000);
                int semester = (int) (Math.random() * 8 + 1);  // 学期为1~8
                Random random = new Random();
                double graduateProbability = 0.7; // 毕业的概率
                String isInSchool = random.nextDouble() < graduateProbability ? "L" : "A";
//                String isInSchool = (Math.random() == 0.8 ? "L" : "A");  // 在籍状态，L为毕业，A在校
                String recordValue = classId + "\t" + name + "\t" + gender + "\t" + birthDateString + "\t"
                        + studentId + "\t" + semester + "\t" + isInSchool;

                // 创建ProducerRecord实例，指定topic
                ProducerRecord<String, String> record = new ProducerRecord<>("stu3", recordValue);

                // 发送消息并获取结果
                try {
                    producer.send(record);
                    System.out.println("Successfully sent message" );
                } catch (Exception e) {
                    e.printStackTrace();
                }

                try {
                    // 等待5秒
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            // 关闭producer


        }



        public static String generateName() {
            List<String> names = new ArrayList<>();
            names.add("Alice");
            names.add("Bob");
            names.add("Charlie");
            names.add("David");
            names.add("Emily");
            names.add("Frank");
            names.add("Grace");
            names.add("Harry");
            names.add("Ivy");
            names.add("John");
            names.add("Katie");
            names.add("Linda");
            names.add("Michael");
            names.add("Nancy");
            names.add("Oliver");
            names.add("Peter");
            names.add("Quinn");
            names.add("Robert");
            names.add("Sophia");
            names.add("Thomas");
            names.add("Ursula");
            names.add("Victoria");
            names.add("William");
            names.add("Xander");
            names.add("Yolanda");
            names.add("Zachary");

            return names.get((int) (Math.random() * names.size()));
        }
    }

